package com.ruoyi.StarStudio.MQTT;

import com.ruoyi.StarStudio.Domain.Device;
import com.ruoyi.StarStudio.Mapper.DeviceMapper;
import com.ruoyi.StarStudio.Mapper.SoilMapper;
import com.ruoyi.common.core.redis.RedisCache;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.sql.Timestamp;

@Component
public class MessageCallBack implements MqttCallback {

    private static final Logger log = LoggerFactory.getLogger(MessageCallBack.class);

    @Autowired
    EmqClient emqClient;

    @Autowired
    RedisCache redisCache;

    @Autowired
    DeviceMapper deviceMapper;

    /**
     * 丢失了对服务端的连接后触发的回调
     * @param throwable
     */
    @Override
    public void connectionLost(Throwable throwable) {
        //资源的清理 重连
        log.info("丢失了与服务端的连接");
        emqClient.reconnect();
        emqClient.subscribe("SoilPH_P", QosEnum.Qos2);
        emqClient.subscribe("Meteorology_P", QosEnum.Qos2);
        emqClient.subscribe("SoilHumiture_P", QosEnum.Qos2);
        emqClient.subscribe("Precipitation_P", QosEnum.Qos2);
        emqClient.subscribe("WindDirection_P", QosEnum.Qos2);
        emqClient.subscribe("WindSpeed_P", QosEnum.Qos2);
    }

    /**
     * 应用收到消息后触发的回调
     * @param topic
     * @param mqttMessage
     * @throws Exception
     */
    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
        log.info("订阅者订阅到了消息，topic={},messageId={},qos={},payload={}",
                topic,mqttMessage.getId(),mqttMessage.getQos(),new String(mqttMessage.getPayload()));
        /*
        * 接下来的操作比较耗时，应该启用新线程进行操作，否则将导致 MQTT连接断开
        */
        new Thread(()->{
            String s = new String(mqttMessage.getPayload());
            if (topic.equals("SoilPH_P")){
                /*
                * 设备序列号：bb57f109-ac28-4762-a614-f798017730d7，土壤PH：7.4
                * */
                //将数据缓存到 Redis中
                log.info("进入方法");
                String[] strings = s.split("，");
                String[] SoilPHSensor_SN_Strings = strings[0].split("：");
                String[] SoilPHStrings = strings[1].split("：");
                int zoneIDByDevice_series_number = deviceMapper.getZoneIDByDevice_series_number(SoilPHSensor_SN_Strings[1]);
                //redisCache.setCacheObject("SoilPHSensor_SN_" + zoneIDByDevice_series_number,SoilPHSensor_SN_Strings[1]);
                redisCache.setCacheObject("soil_ph_" + zoneIDByDevice_series_number, Double.valueOf(SoilPHStrings[1]));
                //将最新数据存入设备表
                Device bySeriesNumber = deviceMapper.getBySeriesNumber(SoilPHSensor_SN_Strings[1]);
                bySeriesNumber.setDevice_data(s.split("，",2)[1]);
                bySeriesNumber.setDevice_updateTime(new Timestamp(System.currentTimeMillis()));
                deviceMapper.update(bySeriesNumber);
            }else if (topic.equals("Meteorology_P")){
                /*
                * 设备序列号：5a5c9617-1682-480d-9d73-475506b3d23b，空气温度：28.7，空气湿度：59.7，气压：6553.5，光照强度：3.9125279E7，CO2浓度：65535.0
                * */
                //将数据缓存到 Redis中
                String[] strings = s.split("，");
                String[] MeteorologySensor_SN_Strings = strings[0].split("：");
                String[] AirTemperatureStrings = strings[1].split("：");
                String[] AirHumidityStrings = strings[2].split("：");
                String[] IlluminanceStrings = strings[4].split("：");
                String[] CO2Strings = strings[5].split("：");
                int zoneIDByDevice_series_number = deviceMapper.getZoneIDByDevice_series_number(MeteorologySensor_SN_Strings[1]);
                //redisCache.setCacheObject("MeteorologySensor_SN_"+zoneIDByDevice_series_number,MeteorologySensor_SN_Strings[1]);
                redisCache.setCacheObject("AirTemperature_"+zoneIDByDevice_series_number,Double.valueOf(AirTemperatureStrings[1]));
                redisCache.setCacheObject("AirHumidity_"+zoneIDByDevice_series_number,Double.valueOf(AirHumidityStrings[1]));
                redisCache.setCacheObject("Illuminance_"+zoneIDByDevice_series_number, Double.valueOf(IlluminanceStrings[1]));
                redisCache.setCacheObject("CO2_"+zoneIDByDevice_series_number, Double.valueOf(CO2Strings[1]));
                //将最新数据存入设备表
                Device bySeriesNumber = deviceMapper.getBySeriesNumber(MeteorologySensor_SN_Strings[1]);
                bySeriesNumber.setDevice_data(s.split("，",2)[1]);
                bySeriesNumber.setDevice_updateTime(new Timestamp(System.currentTimeMillis()));
                deviceMapper.update(bySeriesNumber);
            }else if (topic.equals("SoilHumiture_P")){
                /*
                * Topic数据
                * 设备序列号：33ae79fc-2be0-4842-87b2-f13f9deae591，土壤温度：28.9，土壤湿度：0.0
                * */
                //将数据缓存到 Redis中
                String[] strings = s.split("，");
                String[] SoilHumitureSensor_SN_Strings = strings[0].split("：");
                String[] Soil_TemperatureStrings = strings[1].split("：");
                String[] Soil_Humidity = strings[2].split("：");
                int zoneIDByDevice_series_number = deviceMapper.getZoneIDByDevice_series_number(SoilHumitureSensor_SN_Strings[1]);
                //redisCache.setCacheObject("SoilHumitureSensor_SN_"+zoneIDByDevice_series_number,SoilHumitureSensor_SN_Strings[1]);
                redisCache.setCacheObject("soil_temperature_"+zoneIDByDevice_series_number, Double.valueOf(Soil_TemperatureStrings[1]));
                redisCache.setCacheObject("soil_humidity_"+zoneIDByDevice_series_number, Double.valueOf(Soil_Humidity[1]));
                //将最新数据存入设备表
                Device bySeriesNumber = deviceMapper.getBySeriesNumber(SoilHumitureSensor_SN_Strings[1]);
                bySeriesNumber.setDevice_data(s.split("，",2)[1]);
                bySeriesNumber.setDevice_updateTime(new Timestamp(System.currentTimeMillis()));
                deviceMapper.update(bySeriesNumber);
            }else if (topic.equals("Precipitation_P")){
                /*
                * 设备序列号：75aa487f-6996-4165-8956-8ce36f4db6c1，降水量：0.2
                */
                //将数据缓存到 Redis中
                String[] strings = s.split("，");
                String[] PrecipitationSensor_SN_Strings = strings[0].split("：");
                String[] PrecipitationStrings = strings[1].split("：");
                int zoneIDByDevice_series_number = deviceMapper.getZoneIDByDevice_series_number(PrecipitationSensor_SN_Strings[1]);
                //redisCache.setCacheObject("PrecipitationSensor_SN_"+zoneIDByDevice_series_number, PrecipitationSensor_SN_Strings[1]);
                redisCache.setCacheObject("Precipitation_"+zoneIDByDevice_series_number, Double.valueOf(PrecipitationStrings[1]));
                //将最新数据存入设备表
                Device bySeriesNumber = deviceMapper.getBySeriesNumber(PrecipitationSensor_SN_Strings[1]);
                bySeriesNumber.setDevice_data(s.split("，",2)[1]);
                bySeriesNumber.setDevice_updateTime(new Timestamp(System.currentTimeMillis()));
                deviceMapper.update(bySeriesNumber);
            } else if (topic.equals("WindDirection_P")) {
                /*
                    设备序列号：c9e09dd0-1c4d-40f5-90c6-bf404717f757，风向：315.0
                */
                //将数据缓存到 Redis中
                String[] strings = s.split("，");
                String[] WindDirectionSensor_SN_Strings = strings[0].split("：");
                String[] WindDirectionStrings = strings[1].split("：");
                int zoneIDByDevice_series_number = deviceMapper.getZoneIDByDevice_series_number(WindDirectionSensor_SN_Strings[1]);
                redisCache.setCacheObject("WinDir_" + zoneIDByDevice_series_number, Double.valueOf(WindDirectionStrings[1]));
                //将最新数据存入设备表
                Device bySeriesNumber = deviceMapper.getBySeriesNumber(WindDirectionSensor_SN_Strings[1]);
                bySeriesNumber.setDevice_data(strings[1]);
                bySeriesNumber.setDevice_updateTime(new Timestamp(System.currentTimeMillis()));
                deviceMapper.update(bySeriesNumber);
            }else if (topic.equals("WindSpeed_P")){
                /*
                    设备序列号：e7e53546-0194-4071-9239-8bdac1888b76，风速:6 （m/s）
                */
                //将数据缓存到Redis中
                String[] strings = s.split("，");
                int zoneIDByDevice_series_number = deviceMapper.getZoneIDByDevice_series_number(strings[0].split("：")[1]);
                redisCache.setCacheObject("WindSpeed_" + zoneIDByDevice_series_number,Double.valueOf(strings[1].split("：")[1]));
                //将最新数据存入设备表
                Device bySeriesNumber = deviceMapper.getBySeriesNumber(strings[0].split("：")[1]);
                bySeriesNumber.setDevice_data(strings[1]);
                bySeriesNumber.setDevice_updateTime(new Timestamp(System.currentTimeMillis()));
                deviceMapper.update(bySeriesNumber);
            }
        }).start();
    }

    /**
     * 消息发布者消息发布完成触发的回调
     * @param iMqttDeliveryToken
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        int messageId = iMqttDeliveryToken.getMessageId();
        String[] topics = iMqttDeliveryToken.getTopics();
        log.info("消息发布完成，messageid={}，topics={}",messageId,topics);
    }
}
